package com.test.test3;

import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Reducer;

import java.io.IOException;
import java.math.BigDecimal;
import java.math.RoundingMode;

public class Reducer3 extends Reducer<LongWritable, Text, Text, Text> {

    final Text k = new Text();
    final Text v = new Text();
    // 昨日收盘价
    BigDecimal close = BigDecimal.ZERO;
    // 今日收盘价
    BigDecimal todayClose = BigDecimal.ZERO;

    @Override
    protected void reduce(LongWritable key, Iterable<Text> values, Context context) throws IOException, InterruptedException {
        for (Text value : values) {
            String line = value.toString();
            String[] split = line.split(",");
            String date = split[0];
            // 设置今日收盘价
            todayClose = new BigDecimal(split[1]);
            // 跳过第一天，因为第一天没有昨日收盘价
            if (key.get() == 1) {
                close = new BigDecimal(split[1]);
                break;
            }
            k.set(date + ",");
            // 计算成交量变化百分比每日收益率保留8位小数
            String result = todayClose.subtract(close)
                    .divide(close, 8, RoundingMode.HALF_UP)
                    .multiply(new BigDecimal(100)).toString();
            v.set(String.format("%s", result + "%"));
            // 更新昨日收盘价
            close = todayClose;
            context.write(k, v);
        }
    }
}
